package com.tgy.kafka.Use;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.tgy.kafka.config.KafkaConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Lazy;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;
import java.util.List;
import java.util.Map;
import java.util.Optional;

/***
 * @ClassName: UserKafka
 * @Description: 测试使用kafka
 * @Auther: tanggy3
 * @Date: 11:13 2022/6/25
 * @version : V1.0
 */
@Component
@Slf4j
@Configuration
public class UseKafka {


    /**
     * 监听
     */
    @KafkaListener(topics = "#{'${tgy.kafka.change-consumer.topic}'.split(',')}",
            errorHandler = "consumerAwareErrorHandler", containerFactory = "testChangeBatchFactory")
    public void consumerListener(List<ConsumerRecord<?, ?>> recordLists) {
        log.info("从kafka接收AI任务状态变更数据数量：" + recordLists.size());
        consumer(recordLists);
    }

    /**
     * 处理
     */
    private void consumer(List<ConsumerRecord<?, ?>> recordLists) {
        recordLists.forEach(msgObjcet -> {
            String key = (String) msgObjcet.key();
            Optional<?> kafkaMessage = Optional.ofNullable(msgObjcet.value());
            if (kafkaMessage.isPresent()) {
                Object message = kafkaMessage.get();
                try {
                    String msg = String.valueOf(message);
                    handle(key, msg);
                } catch (Exception e) {
                    log.error("consumer error:", e);
                }
            }
        });
    }


    public void handle(String kafkaKey, String msg) {
        try {
            JSONObject resJson = JSON.parseObject(msg);
            log.info("开始处理AI任务状态变更结果:" + resJson);

            if (resJson != null) {
                log.info("收到Kafka消息===》{}", resJson);
            }
        } catch (Exception e) {
            log.error("The realTimeResult kafka message deals fail,kafkaKey:" + kafkaKey + ",msg:" + msg, e);
        }
    }
}
